package com.bigdata.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * @author Gerry chan
 * @version 1.0
 * DStream join操作：https://www.bilibili.com/video/BV11A411L7CK?p=195
 * 2020/12/27 20：22 周日
 */
object SparkStreaming06_state_Join {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]")
      .setAppName("join")
    val ssc = new StreamingContext(sparkConf, Seconds(3))

    val data999 = ssc.socketTextStream("localhost", 9999)
    val data888 = ssc.socketTextStream("localhost", 8888)

    val map999:DStream[(String, Int)] = data999.flatMap(_.split(" ")).map((_,1))
    val map888:DStream[(String,Int)] = data888.flatMap(_.split(" ")).map((_,1))

    //所谓DStream的Join操作，其实就是两个RDD的join
    val joinDS:DStream[(String, (Int,Int))] = map999.join(map888)
    ssc.start()
    ssc.awaitTermination()
  }
}
